package com.yonyou.common.service.messagecenter.rabbitmq;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;

import org.apache.ibatis.session.RowBounds;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.transaction.annotation.Transactional;

import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.yonyou.common.service.messagecenter.config.RabbitMQConfig;
import com.yonyou.common.service.messagecenter.dao.TtMessageHistoryConsumerMapper;
import com.yonyou.common.service.messagecenter.dao.TtMessageHistoryMapper;
import com.yonyou.common.service.messagecenter.dao.TtResendMessageMapper;
import com.yonyou.common.service.messagecenter.datamapping.DataMapping;
import com.yonyou.common.service.messagecenter.rabbitmq.entity.MessageProtocol;
import com.yonyou.common.service.messagecenter.rabbitmq.entity.MessageProtocolDto;
import com.yonyou.common.service.messagecenter.rabbitmq.entity.TtMessageHistory;
import com.yonyou.common.service.messagecenter.rabbitmq.entity.TtMessageHistoryConsumer;
import com.yonyou.common.service.messagecenter.rabbitmq.entity.TtResendMessage;
import com.yonyou.common.service.messagecenter.rabbitmq.service.AppStatusService;
import com.yonyou.common.service.messagecenter.rabbitmq.service.MessageHistoryConsumerService;
import com.yonyou.common.service.messagecenter.rabbitmq.service.MessageHistoryService;
import com.yonyou.common.service.messagecenter.rabbitmq.service.MessageProtocolService;
import com.yonyou.common.service.messagecenter.rabbitmq.service.ResendMessageService;
import com.yonyou.common.service.messagecenter.util.MqConsts;

import tk.mybatis.mapper.entity.Example;

public abstract class MessageConsumer extends EndPoint implements  Runnable,Consumer{

	private static ConcurrentHashMap<String,String> appTrying=new ConcurrentHashMap<String,String>(); 
	private volatile MessageConverter jackson2JsonMessageConverter=new Jackson2JsonMessageConverter();
	private volatile MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();
    @Autowired
    private RabbitMQConfig config;
	@Autowired
    private MessageProducer messageProducer;
	@Autowired
	private MessageHistoryConsumerService messageHistoryConsumerService;
	@Autowired
	private ResendMessageService resendMessageService;
	@Autowired
	private MessageProtocolService messageProtocolService;
	@Autowired
	private MessageHistoryService messageHistoryService;
	@Autowired
	private TtMessageHistoryMapper historyMapper;
	private DataMapping dataMapping;
	
	
	private Logger logger = LoggerFactory.getLogger(this.getClass());
		
	public MessageConsumer(String endPointName) throws IOException, TimeoutException{
        super(endPointName);       
    }
	public abstract boolean listen(String protocol,Map<String,String> result);
	
	
     
    public void run() {
        try {
        	//设置一次消费1个消息
        	this.getChannel().basicQos(1);
        	boolean autoAck=false;
        	this.getChannel().basicConsume(endPointName, autoAck,this);
        } catch (IOException e) {
        	logger.error("消费失败",e);
            //e.printStackTrace();
        }
    }
 
    /**
     * Called when consumer is registered.
     */
    public void handleConsumeOk(String consumerTag) {
    	logger.info("Consumer---------- "+consumerTag +" registered");
    }
    /** useful code
    	//org.springframework.amqp.core.MessageProperties prop=messagePropertiesConverter.toMessageProperties(props, env, "UTF-8");
		//Message message=new Message(body, prop);
		//Object fo=jackson2JsonMessageConverter.fromMessage(message);
		//MessageProtocol mp=(MessageProtocol)fo;
		
    	//重新进入队列
        //channel.basiNack(map,false,true);
     */
    
    
    public void handleDelivery(String consumerTag, Envelope env,
            BasicProperties props, byte[] body) throws IOException {
    	String odata = new String(body, "UTF-8");
    	MessageProtocolDto mes=messageProtocolService.parseConsumerData(body,this.endPointName);
		try{
	    	logger.info("消息转换完成，消息内容："+mes.toString());
	    	
			//if(!AppStatusService.isErrorApp(mes.getAppid())){
	        int result=this.dealMessage(mes, env,odata);
/*	        	if(result==0)
	        		AppStatusService.addErrorApp(mes.getAppid());
			}else{//发现该app发送失败中,则写入重发表中
				TtMessageHistoryConsumer record=messageHistoryConsumerService.newMessageHistory(mes);
				record.setStatus(MqConsts.CONSUMER_STATUS_DIRECT_FAIL);
				messageHistoryConsumerService.insertHistory(record);
	        	this.changeToReSend(record, odata, env);
			}*/
			TtMessageHistory record =messageHistoryService.getMessageHistory(mes);
			historyMapper.insertLost(record);
		}catch(Exception e){
			logger.error("消息处理失败",e);
			//e.printStackTrace();
			this.getChannel().basicReject(env.getDeliveryTag(), true);
		}finally{
		}
		
		/*		String appStatusKey=mes.getAppid();//代表app接口不可用
		String mesgKey=mes.getAppid()+"try";//代表正在失败重试的key
		String mesValue=mes.getSign()+this.endPointName;//代表处于失败重试中的消息
*/
    }
    
    private int dealMessage(MessageProtocolDto mes, Envelope env,String odata) throws IOException{
    	TtMessageHistoryConsumer record=null;
    	int result=1;
    	try{
    		record=messageHistoryConsumerService.findSuccessMessageHistory(mes.getSign(),this.endPointName);
			if(record==null){
				record=messageHistoryConsumerService.newMessageHistory(mes);
				messageHistoryConsumerService.insertHistory(record);
			}else if(record.getStatus().equals(MqConsts.CONSUMER_STATUS_FINISH)){ 
				this.resendMessageService.updateResendDirectSuccessMessage(mes.getSign(),this.endPointName);
				//重发消息成功，但basicAck失败，会导致增加同样的消息，因此如果处理成功，直接跳出。
				//如果重新发送成功，则确认掉该消息，处理下一条消息。
	        	this.getChannel().basicAck(env.getDeliveryTag(), false);
	        	return 1;
			}
			Map<String,String> map=new HashMap<String,String>();
			boolean isSuccess=this.listen(mes.getData(),map);
	        if(isSuccess){
				this.resendMessageService.updateResendSuccessMessage(mes.getSign(),this.endPointName);
				this.messageHistoryConsumerService.updateMessageHis(record,MqConsts.CONSUMER_STATUS_FINISH,"");
	        	//确认消息收到的应答
	        	this.getChannel().basicAck(env.getDeliveryTag(), false);
	        }else{
	        	messageHistoryConsumerService.updateMessageHis(record,MqConsts.CONSUMER_STATUS_FAIL,map.get(MqConsts.RESPONSE_DATA_MESSAGE));
	        	this.changeToReSend(record, odata, env);
	        	result=0;
	        }
		
		}catch(Exception e){//发生异常则拒绝确认该消息
			logger.error("消息处理失败",e);
			//e.printStackTrace();
            this.getChannel().basicReject(env.getDeliveryTag(), true);
            messageHistoryConsumerService.updateMessageHis(record,MqConsts.CONSUMER_STATUS_FAIL,e.getMessage());
		}
    	return result;
/*	        	//Thread.sleep(1000*config.getPauseSec());//暂停60秒
			messageHistoryConsumerService.updateMessageHis(record,MqConsts.CONSUMER_STATUS_FAIL);
    	boolean r=resendMessageService.insertResendMessage(record,odata);
    	if(r){//如果写入重发表成功，则确认掉该消息，处理下一条消息。
        	this.getChannel().basicAck(env.getDeliveryTag(), false);
		}else{//如果写入失败，则不能拒绝确认该消息
            this.getChannel().basicReject(env.getDeliveryTag(), true);
		}
		result= 1;
		boolean r;
		if((record.getTryCount()%10)==0){//插入消息重发表中
			r=resendMessageService.insertResendMessage(record,odata);
			result= 2;
		}else{//重新发送该消息到本队列，到队列尾部
    		r=messageProducer.sendMessageSimple(MqConsts.DEFAULT_DIRECT_EXCHANGE, this.endPointName, 
    				messageProtocolService.getMessageProtocol(mes),true);
		}
		if(r){//如果重新发送或写入重发表成功，则确认掉该消息，处理下一条消息。
        	this.getChannel().basicAck(env.getDeliveryTag(), false);
		}else{//如果重新发送到本队列失败，则不能拒绝确认该消息
            this.getChannel().basicReject(env.getDeliveryTag(), true);
		}*/
    }
    private boolean changeToReSend(TtMessageHistoryConsumer record,String odata, Envelope env) throws IOException{
    	boolean r=resendMessageService.insertResendMessage(record,odata);
    	if(r){//如果写入重发表成功，则确认掉该消息，处理下一条消息。
        	this.getChannel().basicAck(env.getDeliveryTag(), false);
		}else{//如果写入失败，则不能拒绝确认该消息
            this.getChannel().basicReject(env.getDeliveryTag(), true);
		}
    	return true;
    }
 
    public void handleCancel(String consumerTag) {
    	logger.info("Consumer---------handleCancel- ");  
        }
    public void handleCancelOk(String consumerTag) {

    	logger.info("Consumer---------handleCancelOk- ");  
    }
    public void handleRecoverOk(String consumerTag) {

    	logger.info("Consumer---------handleRecoverOk- ");  
    }
    public void handleShutdownSignal(String consumerTag, ShutdownSignalException arg1) {

    	logger.info("Consumer---------handleShutdownSignal- ");  
    }
	public DataMapping getDataMapping() {
		return dataMapping;
	}
	public void setDataMapping(DataMapping dataMapping) {
		this.dataMapping = dataMapping;
	}
    
}
